package com.example.kafka.service;

import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 演示异步处理同步提交位点的错误示例
 */
@Service
@RequiredArgsConstructor
public class AsyncProcessSyncCommitService {
    private final Logger log = LoggerFactory.getLogger(getClass());

    private final KafkaConsumer<String, String> consumer;
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    /**
     * 用例1: 异步订单处理 - 同步提交位点
     * 错误示例：异步处理订单但同步提交位点
     */
    public String processAsyncOrder(String orderData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：异步处理订单
                CompletableFuture.runAsync(() -> {
                    try {
                        processOrderLogic(record.value());
                    } catch (Exception e) {
                        log.error("Error processing order: {}", record.value(), e);
                    }
                }, executorService);

                // 错误：同步提交位点，不等待异步处理完成
                consumer.commitSync();
            }
            return "Async order processed";
        } catch (Exception e) {
            log.error("Error processing async order", e);
            throw new RuntimeException("Failed to process async order", e);
        }
    }

    /**
     * 用例2: 异步支付处理 - 同步提交位点
     * 错误示例：异步处理支付但同步提交位点
     */
    public String processAsyncPayment(String paymentData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：异步处理支付
                CompletableFuture.runAsync(() -> {
                    try {
                        processPaymentLogic(record.value());
                    } catch (Exception e) {
                        log.error("Error processing payment: {}", record.value(), e);
                    }
                }, executorService);

                // 错误：同步提交位点，不等待异步处理完成
                consumer.commitSync();
            }
            return "Async payment processed";
        } catch (Exception e) {
            log.error("Error processing async payment", e);
            throw new RuntimeException("Failed to process async payment", e);
        }
    }

    /**
     * 用例3: 异步库存更新 - 同步提交位点
     * 错误示例：异步更新库存但同步提交位点
     */
    public String updateAsyncInventory(String inventoryData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：异步更新库存
                CompletableFuture.runAsync(() -> {
                    try {
                        updateInventoryLogic(record.value());
                    } catch (Exception e) {
                        log.error("Error updating inventory: {}", record.value(), e);
                    }
                }, executorService);

                // 错误：同步提交位点，不等待异步处理完成
                consumer.commitSync();
            }
            return "Async inventory updated";
        } catch (Exception e) {
            log.error("Error updating async inventory", e);
            throw new RuntimeException("Failed to update async inventory", e);
        }
    }

    /**
     * 用例4: 异步用户注册 - 同步提交位点
     * 错误示例：异步处理用户注册但同步提交位点
     */
    public String processAsyncUserRegistration(String userData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：异步处理用户注册
                CompletableFuture.runAsync(() -> {
                    try {
                        processUserRegistrationLogic(record.value());
                    } catch (Exception e) {
                        log.error("Error processing user registration: {}", record.value(), e);
                    }
                }, executorService);

                // 错误：同步提交位点，不等待异步处理完成
                consumer.commitSync();
            }
            return "Async user registration processed";
        } catch (Exception e) {
            log.error("Error processing async user registration", e);
            throw new RuntimeException("Failed to process async user registration", e);
        }
    }

    /**
     * 用例5: 异步日志处理 - 同步提交位点
     * 错误示例：异步处理日志但同步提交位点
     */
    public String processAsyncLog(String logData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：异步处理日志
                CompletableFuture.runAsync(() -> {
                    try {
                        processLogLogic(record.value());
                    } catch (Exception e) {
                        log.error("Error processing log: {}", record.value(), e);
                    }
                }, executorService);

                // 错误：同步提交位点，不等待异步处理完成
                consumer.commitSync();
            }
            return "Async log processed";
        } catch (Exception e) {
            log.error("Error processing async log", e);
            throw new RuntimeException("Failed to process async log", e);
        }
    }

    // 模拟业务处理逻辑
    private void processOrderLogic(String orderData) {
        // 实际订单处理逻辑
    }

    private void processPaymentLogic(String paymentData) {
        // 实际支付处理逻辑
    }

    private void updateInventoryLogic(String inventoryData) {
        // 实际库存更新逻辑
    }

    private void processUserRegistrationLogic(String userData) {
        // 实际用户注册处理逻辑
    }

    private void processLogLogic(String logData) {
        // 实际日志处理逻辑
    }
}
